package api.ws.server;

import api.ws.server.msg.MessageExecutorContainer;
import api.ws.server.msg.impl.EchoMessageExecutor;
import basic.common.Cst;
import basic.common.Global;
import basic.common.dubbo.RemoteUserService;
import basic.common.result.SystemException;
import basic.common.rocket.MqModel;
import basic.common.rocket.MqTaskExecutor;
import basic.common.ws.*;
import basic.webflux.cache.RedisKey;
import com.alibaba.fastjson.JSON;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Objects;

/**
 * @创建人 xiaojiancheng
 * @时间 2020/10/7
 * @描述 $
 */
@Component
public class WebSocketServiceImpl implements WebsocketService {
    @Resource
    UserContainer userContainer;

    @Resource
    MqTaskExecutor executor;

    @Resource
    RedisKey redisKey;

    @DubboReference(check = false)
    private RemoteUserService remoteUserService;

    @Resource
    MessageExecutorContainer messageExecutorContainer;

    @Resource
    EchoMessageExecutor echoMessageExecutor;

    @Override
    public Mono<WsContext> login(WsContext context) {
        MsgUser loginUser = context.getLoginUser();
        String token = loginUser.getUser().getToken();

        if(remoteUserService == null){
            return Mono.error(new SystemException(-1,"用户中心没有上线"));
        }

        return Mono.fromFuture(remoteUserService.getUserByToken(token))
                .flatMap(user -> {
                    loginUser.setUser(user);
                    userContainer.put(loginUser, context.getChannel());
                    return Mono.just(context);
                });

    }

    @Override
    public Mono<WsContext> onMessage(WsContext context) {
        //  如果是 echo 消息 直接回复了
        Message msg = context.getMessage();
        if(Objects.equals(msg.getType(), Cst.MessageType.ECHO)){
            return echoMessageExecutor.execute(context)
                    .then(Mono.just(context));
        }

        MqModel model = new MqModel();
        //  不需要traceId
        model.setTopic(WsCst.TopicName.WS_SEND);
        model.setTag(WsCst.MsgType.SEND_USER);
        model.obj(msg);

        //  异步发送
        return Mono.create(item -> {
            Global.tpe.execute(() -> {
                try{
                    executor.publish(model);
                    item.success();
                }catch (Exception e){
                    item.error(e);
                }

            });
        })
                .then(Mono.just(context));
    }
}
